package com.example.study;

import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxjavaStudy {
    public static void main(String[] args) throws InterruptedException {
//        method1();
//        method2();
        method3();
        Thread.sleep(20000);
    }

    /**
     * 定时任务
     */
    private static void method1() {
        Observable.timer(1000, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        printlnString.printlnString("开始订阅");
                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        printlnString.printlnString("onNext");

                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        printlnString.printlnString("onError");
                    }

                    @Override
                    public void onComplete() {
                        printlnString.printlnString("onComplete");
                    }
                });
    }

    static Disposable disposable;

    //定时间隔任务
    private static void method2() {
        Observable.interval(1000, 1000, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        disposable = d;
                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {

                        if (aLong == 10) {
                            //取消订阅
                            if (disposable != null && disposable.isDisposed()) {
                                disposable.dispose();
                            }
                        }
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        if (disposable != null && disposable.isDisposed()) {
                            disposable.dispose();
                        }
                    }

                    @Override
                    public void onComplete() {
                        if (disposable != null && disposable.isDisposed()) {
                            disposable.dispose();
                        }
                    }
                });
    }

    private static void method3() {

        Observable.just("String1", "String2", "String3", "String4")
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        printlnString.printlnString("开始订阅");
                    }

                    @Override
                    public void onNext(@NonNull String s) {
                        printlnString.printlnString(" 上游的数据onNext =" + s);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        printlnString.printlnString(" Throwable =" + "Throwable");
                    }

                    @Override
                    public void onComplete() {
                        printlnString.printlnString("完成");
                    }
                });
    }

    /**
     * map模式
     */

    static class User {
        private String name;
        private int age;

        public User(String name, int age) {
            this.name = name;
            this.age = age;
        }
    }

    private static void method4() {
        Observable.just(new User("砂糖橘", 30))
                //参数是User 给下游的数据为String
                .map(new Function<User, String>() {
                    @Override
                    public String apply(User user) throws Throwable {
                        return null;
                    }

                })
                //上游传递的数据为String，给到下游的数据为User
                .map(new Function<String, User>() {
                    @Override
                    public User apply(String s) throws Throwable {
                        return null;
                    }
                }).subscribe(new Observer<User>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull User user) {
                // 上游给到的数据
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

}
